package com.imooc.dianping.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.google.common.collect.Lists;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

/**
 * canal 客户端
 *
 * @author gblfy
 * @date 2021-11-22
 */
@Component
public class CanalClient implements DisposableBean {
    private CanalConnector canalConnector;

    @Value("${canal.alone-ip}")
    private String CANALIP;
    @Value("${canal.destination}")
    private String DESTINATION;
    @Value("${canal.username}")
    private String USERNAME;
    @Value("${canal.passwoed}")
    private String PASSWOED;
    @Value("${canal.port}")
    private int PORT;

    @Bean
    public CanalConnector getCanalConnector() {
        //canal实例化
        canalConnector = CanalConnectors.newClusterConnector(Lists.newArrayList(
                new InetSocketAddress(CANALIP, PORT)), DESTINATION, USERNAME, PASSWOED);
        //连接canal
        canalConnector.connect();
        // 自定filter 格式{database}.{table}
        canalConnector.subscribe();
        // 回滚寻找上次中断的位置
        canalConnector.rollback();
        //返回连接
        return canalConnector;
    }

    @Override
    public void destroy() throws Exception {
        if (canalConnector != null) {
            //防止canal泄露
            canalConnector.disconnect();
        }
    }
}
